/*
 * ebcd_pch.c - EBCD PCH
 * Clunix, KyungWoon Cho, cezanne@clunix.com, 2006.2
 *
 * Copyright (c) 2001-2008 by Clunix Inc. All Rights Reserved.
 */

#include "config.h"

#include "ebcd.h"
#include "audit.h"
#include "monitor.h"

ecm_mutex_t	accepted_mutex;
LIST_HEAD(accepted_pchs);

ebc_pch_t	*pchp_accept;
ebc_pch_t	*pchp_bdcast;
ebcd_pch_t	*dpch_bdcast;

ebc_selection_t	*ebcd_selection;

extern void ebcd_init_epchs(void);
extern void ebcd_init_lpchs(void);

static ebc_pktb_t *
build_aliveness_pktb(ebc_ndid_t ndid)
{
	ebc_pktb_t	*pktb;
	ebc_mhdr_t	*mhdr;

	pktb = ebc_pktb_new();
	if (pktb == NULL)
		return NULL;
	pktb->pkthdr.clid = ebcd_clid;

	mhdr = &pktb->pkthdr.mhdr;

	mhdr->msgt = EBC_MSGT_ALIVE | EBC_MSGT_REQ | EBC_MSGT_ONEWAY;
	mhdr->dstaddr.ndid = ndid;
	mhdr->dstaddr.mdid = EBC_MDID_EBCD;
	mhdr->srcaddr.ndid = ebcd_ndid;
	mhdr->srcaddr.mdid = EBC_MDID_EBCD;

	ebc_pkthdr_encode(pktb);

	return pktb;
}

ebcd_pch_t *
create_ebcd_pch(ebc_pch_t *pchp, ebcd_pch_type_t type)
{
	ebcd_pch_t	*dpch;

	dpch = ecm_malloc(sizeof(ebcd_pch_t));
	dpch->type = type;
	dpch->pktb_aliveness = build_aliveness_pktb(pchp->ndid);
	dpch->pchp = pchp;
	dpch->refcnt = 1;
	dpch->mutex = ecm_mutex_create(ECM_FALSE);
	dpch->destroyed = ECM_FALSE;
	dpch->routing = ECM_FALSE;
	dpch->last_sent = 0;
	dpch->last_recv = 0;
	dpch->n_routing_pktbs = 0;
	INIT_LIST_HEAD(&dpch->routing_pktbs);
	INIT_LIST_HEAD(&dpch->list_sent);
	INIT_LIST_HEAD(&dpch->list_recv);

	if (type == EXT_PCH)
		dpch->alivecheck = ECM_TRUE;
	else
		dpch->alivecheck = ECM_FALSE;
	return dpch;
}

static void
free_dpch(ebcd_pch_t *dpch)
{
	ebc_pch_t	*pchp = dpch->pchp;

	EBCDDBG(DBG_PCH, "PCH Removed: %s", ebc_dbg_pch(pchp));

	if (dpch->type == EXT_PCH) {
		ebcd_free_epch(dpch);
		ECMVERBOSE("%s: node disconnected", ecm_safe_str(pchp->name));
	}
	else if (dpch->type == LOCAL_PCH)
		ebcd_free_lpch(dpch);

	ebc_pch_free(pchp);

	ebc_pktb_free(dpch->pktb_aliveness);
	ecm_mutex_free(dpch->mutex);
	ecm_free(dpch);
}

ECMBOOL
ebcd_pch_get(ebcd_pch_t *dpch)
{
	ecm_mutex_lock(dpch->mutex);
	if (dpch->destroyed) {
		ecm_mutex_unlock(dpch->mutex);
		return ECM_FALSE;
	}
	dpch->refcnt++;
	ecm_mutex_unlock(dpch->mutex);
	return ECM_TRUE;
}

void
ebcd_pch_put(ebcd_pch_t *dpch)
{
	if (dpch == NULL)
		return;

	ecm_mutex_lock(dpch->mutex);

	ECMASSERT(dpch->refcnt > 0);
	dpch->refcnt--;
	if (dpch->refcnt == 0 && dpch->destroyed) {
		ecm_mutex_unlock(dpch->mutex);
		free_dpch(dpch);
	}
	else
		ecm_mutex_unlock(dpch->mutex);
}

ebcd_pch_t *
ebcd_add_pch(ebc_pch_t *pchp)
{
	ebcd_pch_t	*dpch;

	if (!EBCD_IS_SELF_NDID(pchp->ndid))
		dpch = ebcd_add_epch(pchp);
	else
		dpch = ebcd_add_lpch(pchp);

#ifdef ECM_DEBUG
	if (dpch)
		EBCDDBG(DBG_PCH, "PCH Added: %s", ebc_dbg_pch(pchp));
#endif

	return dpch;
}

void
ebcd_del_pch(ebcd_pch_t *dpch)
{
	if (dpch == NULL)
		return;
	free_dpch(dpch);
}

void
ebcd_accept_pch(ebcd_pch_t *dpch)
{
	ebc_pch_t	*pchp;

	if (dpch == NULL)
		return;

	pchp = dpch->pchp;

	ecm_mutex_lock(accepted_mutex);

	PCH_SELECTION_ADD(pchp, ebcd_selection);

	pchp->clid = ebcd_clid;
	list_add_tail(&dpch->list, &accepted_pchs);

	if (dpch->alivecheck)
		ebcd_init_alivecheck(dpch);

	ecm_mutex_unlock(accepted_mutex);
	setup_moninfo(dpch, ECM_TRUE);
	ebcd_pch_put(dpch);
}

void
ebcd_destroy_pch(ebcd_pch_t *dpch)
{
	ecm_mutex_lock(accepted_mutex);
	ecm_mutex_lock(dpch->mutex);
	if (dpch->destroyed) {
		ecm_mutex_unlock(dpch->mutex);
		ecm_mutex_unlock(accepted_mutex);
		return;
	}
	dpch->destroyed = ECM_TRUE;
	ecm_mutex_unlock(dpch->mutex);

	EBCDDBG(DBG_DPCH, "destroy: %s", ebcd_dbg_pch(dpch));

	PCH_SELECTION_DEL(dpch->pchp);
	list_del_init(&dpch->list);
	ecm_mutex_unlock(accepted_mutex);

	if (dpch->alivecheck)
		ebcd_clear_alivecheck(dpch);

	setup_moninfo(dpch, ECM_FALSE);

	ebcd_cancel_msg(dpch);
	clear_audit_infos(dpch);
}

static ECM_INLINE ECMBOOL
pktb_has_valid_src(ebc_pktb_t *pktb)
{
	ebc_ndid_t	src_ndid;

	src_ndid = pktb->pkthdr.mhdr.srcaddr.ndid;
	if (src_ndid == ebcd_ndid || ebcd_has_epch(src_ndid))
		return ECM_TRUE;
	return ECM_FALSE;
}

void
ebcd_drop_pktb(ebc_pktb_t *pktb)
{
	ebc_mhdr_t	*mhdr;

	mhdr = &pktb->pkthdr.mhdr;
	if (!(mhdr->msgt & (EBC_MSGT_DIRECT | EBC_MSGT_ONEWAY)))
		ebcd_push_errmsg(&mhdr->dstaddr, &mhdr->srcaddr, mhdr->msgid, EBC_MSGT_ERR_ROUTE, ECM_FALSE);
	EBCDDBG(DBG_PKTDROP, "invalid dest. drop: %s", ebc_dbg_pktb(pktb));
	ebc_pktb_free(pktb);
}

void
process_detection(ebcd_pch_t *dpch, ebc_pktb_t *pktb)
{
	ebc_mhdr_t	*mhdr;
	ebc_addr_t	*paddr_src;

	mhdr = &pktb->pkthdr.mhdr;
	paddr_src = &mhdr->srcaddr;
	if (paddr_src->mdid != EBC_MDID_EBCD ||
	    paddr_src->ndid == ebcd_ndid ||
	    ebcd_has_epch(mhdr->srcaddr.ndid)) {
		ebc_pktb_free(pktb);
		return;
	}
	if (!(mhdr->msgt & EBC_MSGT_REQ)) {
		ebcd_build_route(dpch, pktb);
		return;
	}
	mhdr->msgt = EBC_MSGT_DETECT | EBC_MSGT_ONEWAY;
	mhdr->dstaddr.ndid = mhdr->srcaddr.ndid;
	mhdr->srcaddr.ndid = ebcd_ndid;
	ebc_pkthdr_encode(pktb);
	ebcd_push_pktb(dpch, pktb);
}

static ebcd_pch_t *
get_next_dpch(ebcd_pch_t *dpch)
{
	ebcd_pch_t	*dpch_next;
	struct list_head	*lp;

	for (lp = dpch->list.next; lp != &accepted_pchs; lp = lp->next) {
		dpch_next = list_entry(lp, ebcd_pch_t, list);
		if (ebcd_pch_get(dpch_next))
			return dpch_next;
	}
	return NULL;
}

static void
push_destroying_dpch(ecm_ptrs_t **pptrs, ebcd_pch_t *dpch)
{
	ecm_ptrs_t	*ptrs = *pptrs;

	if (ptrs == NULL)
		*pptrs = ptrs = ecm_ptrs_new(4);
	ecm_ptrs_add(ptrs, dpch);
}

static void
clear_destroying_dpchs(ecm_ptrs_t *ptrs)
{
	ebcd_pch_t	*dpch;
	ecm_uint32	i;

	if (ptrs == NULL)
		return;

	for (i = 0; i < ptrs->len; i++) {
		dpch = (ebcd_pch_t *)ptrs->pdata[i];
		ebcd_destroy_pch(dpch);
		ebcd_pch_put(dpch);
	}
	ecm_ptrs_free(ptrs);
}

void
ebcd_write_pchs(void)
{
	ebcd_pch_t	*dpch, *dpch_next;
	ecm_ptrs_t	*ptrs_destroying = NULL;
	ebc_err_t	err;

	ecm_mutex_lock(accepted_mutex);

	if (list_empty(&accepted_pchs)) {
		ecm_mutex_unlock(accepted_mutex);
		return;
	}

	dpch = list_entry(accepted_pchs.next, ebcd_pch_t, list);
	EBCD_PCH_GET_ASSERT(dpch);
	while (dpch) {
		dpch_next = get_next_dpch(dpch);

		err = pch_write_all(dpch->pchp);

		if (err == -ERR_EBC_DISCONN) {
			EBCDDBG(DBG_DPCH, "%s: disconnected while writing", ebcd_dbg_pch(dpch));
			push_destroying_dpch(&ptrs_destroying, dpch);
		}
		else {
			ebcd_pch_put(dpch);
		}
		dpch = dpch_next;
	}

	ecm_mutex_unlock(accepted_mutex);

	clear_destroying_dpchs(ptrs_destroying);
}

static ECMBOOL
setup_accept_pch(void)
{
	char	*addrspec;

	ebcd_selection = PCH_CREATE_SELECTION(EBC_PCH_SOCK_TYPE);

	ecm_asprintf(&addrspec, "sock/tcp/?.:%hu", ebcd_port);
	pchp_accept = ebc_pch_create(addrspec, -1);
	ecm_free(addrspec);
	if (pchp_accept == NULL)
		return ECM_FALSE;
	pchp_accept->clid = ebcd_clid;

	PCH_SELECTION_ADD(pchp_accept, ebcd_selection);
	return ECM_TRUE;
}

static ECMBOOL
setup_bdcast_pch(void)
{
	char	*addrspec;

	if (ebcd_bdcast_addr == NULL)
		ecm_asprintf(&addrspec, "sock/udp/.:%hu", ebcd_port);
	else
		ecm_asprintf(&addrspec, "sock/udp/%s:%hu", ebcd_bdcast_addr, ebcd_port);
	pchp_bdcast = ebc_pch_create(addrspec, -1);
	ecm_free(addrspec);

	if (pchp_bdcast == NULL)
		return ECM_FALSE;

	pchp_bdcast->name = ecm_strdup("<bdcast>");
	pchp_bdcast->clid = ebcd_clid;
	pchp_bdcast->ndid = ebcd_ndid;

	dpch_bdcast = create_ebcd_pch(pchp_bdcast, BDCAST_PCH);
	return ECM_TRUE;
}

ECMBOOL
ebcd_initialize_pchs(void)
{
	if (ebcd_port == 0)
		ebcd_port = EBC_PORT_DEF;
	ebc_get_ndid(ebcd_ndname, &ebcd_ndid);
	ebc_set_addr(&ebcd_myaddr, ebcd_ndid, EBC_MDID_EBCD, EBC_PTID_DEFAULT);

	accepted_mutex = ecm_mutex_create(ECM_FALSE);

	if (!setup_accept_pch() || !setup_bdcast_pch())
		return ECM_FALSE;

	ebcd_accept_pch(dpch_bdcast);

	ebcd_init_epchs();
	ebcd_init_lpchs();

	return ECM_TRUE;
}

void
ebcd_finalize_pchs(void)
{
	free_dpch(dpch_bdcast);
	PCH_FREE_SELECTION(ebcd_selection);
	ebc_pch_free(pchp_accept);
	ecm_mutex_free(accepted_mutex);
}

#ifdef ECM_DEBUG

static ecm_uint32
get_accepted_count(void)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;
	ecm_uint32	n_accepted = 0;

	ecm_mutex_lock(accepted_mutex);
	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);
		n_accepted++;
	}
	ecm_mutex_unlock(accepted_mutex);

	return n_accepted;
}

sr_t *
report_pch_count(void)
{
	sr_t	*sr_res;
	extern ecm_uint32 get_epchs_count(void);
	extern ecm_uint32 get_lpchs_count(void);

	sr_res = sr_create_dir();

	sr_createc_int(sr_res, "n_accepted", ST_INT, get_accepted_count());
	sr_createc_int(sr_res, "n_epchs", ST_INT, get_epchs_count());
	sr_createc_int(sr_res, "n_lpchs", ST_INT, get_lpchs_count());

	return sr_res;
}

static sr_t *
build_pch_info(ebcd_pch_t *dpch)
{
	ebc_pch_t	*pchp = dpch->pchp;
	sr_t		*sr_res;

	sr_res = sr_create_dir();
	sr_append(sr_res, sr_create_int(ST_INT, pchp->n_pktbs_in_spktq));

	return sr_res;
}

sr_t *
report_pch_info(const char *name)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;
	sr_t	*sr_res = NULL;

	ecm_mutex_lock(accepted_mutex);
	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);

		if (ecm_strcmp(name, ebcd_dbg_pch(dpch)) == 0) {
			sr_res = build_pch_info(dpch);
			break;
		}
	}
	ecm_mutex_unlock(accepted_mutex);

	return sr_res;
}

sr_t *
report_pch_list(void)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;
	sr_t	*sr_res, *sr_pch_info;

	sr_res = sr_create_dir();

	ecm_mutex_lock(accepted_mutex);
	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);
		sr_pch_info = sr_create_dir();
		sr_set_name(sr_pch_info, ebcd_dbg_pch(dpch));
		sr_append(sr_res, sr_pch_info);
	}
	ecm_mutex_unlock(accepted_mutex);

	return sr_res;
}

sr_t *
report_pktb(void)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;
	sr_t	*sr_res;
	ecm_uint32	n_all_pktbs_in_spktq = 0;
	extern ecm_uint32	n_pktbs;

	sr_res = sr_create_dir();

	sr_append(sr_res, sr_create_int(ST_INT, n_pktbs));
	ecm_mutex_lock(accepted_mutex);
	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);
		n_all_pktbs_in_spktq += dpch->pchp->n_pktbs_in_spktq;
	}
	ecm_mutex_unlock(accepted_mutex);

	sr_append(sr_res, sr_create_int(ST_INT, n_all_pktbs_in_spktq));

	return sr_res;
}

sr_t *
report_call_infos(void)
{
	sr_t	*sr_res, *sr_res1;
	ebcd_pch_t	*dpch;
	struct list_head	*lp;
	extern sr_t *report_call_info(ebcd_pch_t *);

	sr_res = sr_create_dir();
	ecm_mutex_lock(accepted_mutex);
	list_for_each(lp, &accepted_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list);
		sr_res1 = report_call_info(dpch);
		sr_set_name(sr_res1, ebcd_dbg_pch(dpch));
		sr_append(sr_res, sr_res1);
	}
	ecm_mutex_unlock(accepted_mutex);

	return sr_res;
}

#endif
